package com.flink.hbase.windowingoptimization;

import com.flink.hbase.windowingoptimization.config.WatermarkConfig;
import com.flink.hbase.windowingoptimization.keyby.KeyByOptimization;
import com.flink.hbase.windowingoptimization.monitoring.WindowingMetricsCollector;
import com.flink.hbase.windowingoptimization.performance.PerformanceTuningGuide;
import com.flink.hbase.windowingoptimization.shuffle.ShuffleOptimization;
import com.flink.hbase.windowingoptimization.watermark.RealTimeAnalyticsExample;
import com.flink.hbase.windowingoptimization.window.AdvancedWindowFunctions;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 窗口优化主作业类
 * 整合所有窗口优化功能模块
 * 
 * 功能模块：
 * 1. Watermark配置和策略
 * 2. 窗口函数高级用法
 * 3. keyBy优化和数据倾斜处理
 * 4. Shuffle分区优化
 * 5. 性能调优最佳实践
 * 6. 监控和告警系统
 * 
 * 使用场景：
 * - 实时数据分析
 * - 流式计算性能优化
 * - 生产环境监控
 * - 数据倾斜处理
 * 
 * @author Flink Developer
 */
public class WindowingOptimizationJob {
    
    private static final Logger logger = LoggerFactory.getLogger(WindowingOptimizationJob.class);
    
    /**
     * 统一的事件数据模型
     */
    public static class UnifiedEvent {
        public String eventId;
        public String key;
        public String value;
        public long timestamp;
        public String category;
        public String eventType;
        public double amount;
        public int size;
        public String source;
        public int priority;
        
        public UnifiedEvent() {}
        
        public UnifiedEvent(String eventId, String key, String value, long timestamp, 
                          String category, String eventType, double amount, int size, 
                          String source, int priority) {
            this.eventId = eventId;
            this.key = key;
            this.value = value;
            this.timestamp = timestamp;
            this.category = category;
            this.eventType = eventType;
            this.amount = amount;
            this.size = size;
            this.source = source;
            this.priority = priority;
        }
        
        @Override
        public String toString() {
            return String.format("UnifiedEvent{eventId='%s', key='%s', category='%s', " +
                    "eventType='%s', timestamp=%d, amount=%.2f, size=%d, source='%s', priority=%d}",
                    eventId, key, category, eventType, timestamp, amount, size, source, priority);
        }
    }
    
    /**
     * 完整的窗口优化示例
     */
    public void runCompleteOptimizationExample() throws Exception {
        logger.info("启动完整的窗口优化示例...");
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 环境优化配置
        configureEnvironment(env);
        
        // 数据源
        DataStream<UnifiedEvent> eventStream = createOptimizedDataSource(env);
        
        // 应用所有优化技术
        SingleOutputStreamOperator<String> optimizedStream = applyAllOptimizations(eventStream);
        
        // 输出结果
        optimizedStream.addSink(new OptimizedSink()).name("optimized-sink");
        
        env.execute("Complete Windowing Optimization Example");
    }
    
    /**
     * 环境配置优化
     */
    private void configureEnvironment(StreamExecutionEnvironment env) {
        // 并行度配置
        env.setParallelism(4);
        env.setMaxParallelism(128);
        
        // 缓冲区优化
        env.setBufferTimeout(10); // 10ms缓冲超时
        
        // 延迟跟踪
        env.getConfig().setLatencyTrackingInterval(1000);
        
        // 重启策略
        env.enableCheckpointing(30000); // 30秒检查点
        
        logger.info("环境配置完成: 并行度={}, 缓冲超时={}ms", 
                   env.getParallelism(), 10);
    }
    
    /**
     * 创建优化的数据源
     */
    private DataStream<UnifiedEvent> createOptimizedDataSource(StreamExecutionEnvironment env) {
        // 方式1：Kafka数据源
        if (System.getProperty("use.kafka.source", "false").equals("true")) {
            return createKafkaSource(env);
        }
        
        // 方式2：模拟数据源
        return env.addSource(new OptimizedEventSource()).name("optimized-event-source");
    }
    
    /**
     * 创建Kafka数据源
     */
    private DataStream<UnifiedEvent> createKafkaSource(StreamExecutionEnvironment env) {
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProps.setProperty("group.id", "windowing-optimization");
        kafkaProps.setProperty("auto.offset.reset", "latest");
        
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
            "windowing-events",
            new SimpleStringSchema(),
            kafkaProps
        );
        
        return env.addSource(kafkaConsumer)
            .map(new KafkaEventParser())
            .name("kafka-source");
    }
    
    /**
     * Kafka事件解析器
     */
    public static class KafkaEventParser implements MapFunction<String, UnifiedEvent> {
        
        @Override
        public UnifiedEvent map(String value) throws Exception {
            // 简化的JSON解析（实际应使用JSON库）
            String[] parts = value.split(",");
            if (parts.length >= 10) {
                return new UnifiedEvent(
                    parts[0].trim(),  // eventId
                    parts[1].trim(),  // key
                    parts[2].trim(),  // value
                    Long.parseLong(parts[3].trim()),  // timestamp
                    parts[4].trim(),  // category
                    parts[5].trim(),  // eventType
                    Double.parseDouble(parts[6].trim()),  // amount
                    Integer.parseInt(parts[7].trim()),  // size
                    parts[8].trim(),  // source
                    Integer.parseInt(parts[9].trim())   // priority
                );
            }
            return null;
        }
    }
    
    /**
     * 应用所有优化技术
     */
    private SingleOutputStreamOperator<String> applyAllOptimizations(DataStream<UnifiedEvent> eventStream) {
        // 1. 指标收集
        SingleOutputStreamOperator<UnifiedEvent> monitoredStream = eventStream
            .process(new WindowingMetricsCollector.MetricsCollector())
            .name("metrics-collection");
        
        // 2. Watermark策略
        SingleOutputStreamOperator<UnifiedEvent> watermarkedStream = monitoredStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.adaptiveStrategy(
                    (event, recordTimestamp) -> event.timestamp,
                    Duration.ofSeconds(10),
                    Duration.ofMinutes(1)
                )
            )
            .name("watermark-assignment");
        
        // 3. 智能分区
        SingleOutputStreamOperator<UnifiedEvent> partitionedStream = watermarkedStream
            .partitionCustom(new ShuffleOptimization.SmartRandomPartitioner(), 
                           event -> event.key)
            .name("smart-partitioning");
        
        // 4. 数据倾斜处理
        SingleOutputStreamOperator<UnifiedEvent> skewHandledStream = partitionedStream
            .process(new KeyByOptimization.HotDataProcessor())
            .name("skew-handling");
        
        // 5. 高级窗口处理
        SingleOutputStreamOperator<AdvancedWindowFunctions.AdvancedTransactionStats> windowedStream = 
            skewHandledStream
                .map(new EventToTransactionConverter())
                .keyBy(event -> event.category)
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .aggregate(
                    new AdvancedWindowFunctions.AdvancedTransactionAggregateFunction(),
                    new AdvancedWindowFunctions.IncrementalWithFullWindowFunction()
                )
                .name("advanced-windowing");
        
        // 6. 性能监控
        SingleOutputStreamOperator<String> monitoredResults = windowedStream
            .map(new WindowingMetricsCollector.MonitoringAggregator())
            .name("performance-monitoring");
        
        return monitoredResults;
    }
    
    /**
     * 事件转换器
     */
    public static class EventToTransactionConverter 
            implements MapFunction<UnifiedEvent, AdvancedWindowFunctions.TransactionEvent> {
        
        @Override
        public AdvancedWindowFunctions.TransactionEvent map(UnifiedEvent event) throws Exception {
            return new AdvancedWindowFunctions.TransactionEvent(
                event.eventId,
                event.key,
                "merchant_" + (event.hashCode() % 100),
                event.category,
                event.amount,
                "CNY",
                event.timestamp,
                event.priority > 7 ? "HIGH" : (event.priority > 4 ? "MEDIUM" : "LOW"),
                event.source,
                "location_" + (event.hashCode() % 10)
            );
        }
    }
    
    /**
     * 优化的数据源
     */
    public static class OptimizedEventSource implements SourceFunction<UnifiedEvent> {
        
        private volatile boolean running = true;
        private final ThreadLocalRandom random = ThreadLocalRandom.current();
        
        @Override
        public void run(SourceContext<UnifiedEvent> ctx) throws Exception {
            String[] categories = {"电商", "金融", "游戏", "社交", "新闻", "教育", "医疗", "交通"};
            String[] eventTypes = {"click", "view", "purchase", "login", "logout", "search", "share"};
            String[] sources = {"web", "mobile", "api", "batch", "stream"};
            
            long eventId = 0;
            
            while (running) {
                // 生成批量事件以提高性能
                for (int i = 0; i < 100; i++) {
                    String key = generateKey(eventId);
                    
                    UnifiedEvent event = new UnifiedEvent(
                        "evt_" + eventId++,
                        key,
                        "data_" + random.nextInt(1000),
                        System.currentTimeMillis(),
                        categories[random.nextInt(categories.length)],
                        eventTypes[random.nextInt(eventTypes.length)],
                        random.nextDouble() * 1000,
                        random.nextInt(1000) + 100,
                        sources[random.nextInt(sources.length)],
                        random.nextInt(10) + 1
                    );
                    
                    ctx.collect(event);
                }
                
                Thread.sleep(100); // 100ms间隔
            }
        }
        
        private String generateKey(long eventId) {
            // 模拟数据倾斜：某些key出现频率更高
            if (random.nextDouble() < 0.3) {
                return "hot_key_" + (eventId % 5); // 热点key
            } else if (random.nextDouble() < 0.6) {
                return "medium_key_" + (eventId % 50); // 中频key
            } else {
                return "normal_key_" + (eventId % 500); // 正常key
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 优化的输出Sink
     */
    public static class OptimizedSink implements SinkFunction<String> {
        
        private long processedCount = 0;
        private long lastReportTime = System.currentTimeMillis();
        
        @Override
        public void invoke(String value, Context context) throws Exception {
            processedCount++;
            
            // 定期输出处理统计
            long currentTime = System.currentTimeMillis();
            if (currentTime - lastReportTime > 30000) { // 30秒
                logger.info("处理统计: 总计处理 {} 条记录", processedCount);
                lastReportTime = currentTime;
            }
            
            // 在这里可以输出到各种存储系统
            // 如HBase、Elasticsearch、Kafka等
            if (processedCount % 1000 == 0) {
                logger.debug("处理结果: {}", value);
            }
        }
    }
    
    /**
     * 性能测试模式
     */
    public void runPerformanceTest() throws Exception {
        logger.info("启动性能测试模式...");
        
        PerformanceTuningGuide.PerformanceBenchmark benchmark = 
            new PerformanceTuningGuide.PerformanceBenchmark();
        
        // 测试不同的优化策略
        benchmark.runBenchmark("baseline", () -> {
            try {
                runBasicWindowingExample();
            } catch (Exception e) {
                logger.error("基线测试失败", e);
            }
        });
        
        benchmark.runBenchmark("optimized", () -> {
            try {
                runCompleteOptimizationExample();
            } catch (Exception e) {
                logger.error("优化测试失败", e);
            }
        });
        
        benchmark.printBenchmarkResults();
    }
    
    /**
     * 基础窗口示例（作为基线）
     */
    public void runBasicWindowingExample() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        
        DataStream<UnifiedEvent> eventStream = env.addSource(new OptimizedEventSource());
        
        SingleOutputStreamOperator<Tuple2<String, Long>> basicResult = eventStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<UnifiedEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((event, recordTimestamp) -> event.timestamp)
            )
            .keyBy(event -> event.category)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .reduce((e1, e2) -> {
                return new UnifiedEvent(
                    e1.eventId,
                    e1.key,
                    e1.value,
                    Math.max(e1.timestamp, e2.timestamp),
                    e1.category,
                    e1.eventType,
                    e1.amount + e2.amount,
                    e1.size + e2.size,
                    e1.source,
                    e1.priority
                );
            })
            .map(event -> new Tuple2<>(event.category, 1L))
            .name("basic-windowing");
        
        basicResult.print("基础窗口结果");
        
        env.execute("Basic Windowing Example");
    }
    
    /**
     * 示例选择器
     */
    public void runExample(String exampleType) throws Exception {
        logger.info("运行示例: {}", exampleType);
        
        switch (exampleType.toLowerCase()) {
            case "complete":
                runCompleteOptimizationExample();
                break;
            case "basic":
                runBasicWindowingExample();
                break;
            case "performance":
                runPerformanceTest();
                break;
            case "realtime":
                new RealTimeAnalyticsExample().runRealTimeAnalytics();
                break;
            case "shuffle":
                new ShuffleOptimization().runBufferOptimization();
                break;
            case "keyby":
                new KeyByOptimization().runDynamicLoadBalancing();
                break;
            case "window":
                new AdvancedWindowFunctions().runMultiLayerWindowAnalysis();
                break;
            default:
                logger.error("不支持的示例类型: {}", exampleType);
                System.out.println("支持的示例类型:");
                System.out.println("  complete - 完整优化示例");
                System.out.println("  basic - 基础窗口示例");
                System.out.println("  performance - 性能测试");
                System.out.println("  realtime - 实时分析");
                System.out.println("  shuffle - 分区优化");
                System.out.println("  keyby - keyBy优化");
                System.out.println("  window - 高级窗口函数");
        }
    }
    
    /**
     * 主函数
     */
    public static void main(String[] args) throws Exception {
        WindowingOptimizationJob job = new WindowingOptimizationJob();
        
        String exampleType = args.length > 0 ? args[0] : "complete";
        
        logger.info("窗口优化作业启动...");
        logger.info("示例类型: {}", exampleType);
        
        try {
            job.runExample(exampleType);
        } catch (Exception e) {
            logger.error("作业执行失败", e);
            System.exit(1);
        }
        
        logger.info("窗口优化作业完成");
    }
} 